package aggregation

import (
	"context"
	"fmt"
	"io"
	"runtime"
	"strings"
	"time"

	"gitee.com/bjf-fhe/apicat/entry"
	"gitee.com/bjf-fhe/apicat/platform"
	"github.com/denisbrodbeck/machineid"
	"github.com/sirupsen/logrus"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/keepalive"
	"google.golang.org/grpc/status"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"

	"github.com/pkg/errors"
)

type GrpcClient struct {
	// daemon      GrpcDaemon
	conn    *grpc.ClientConn
	localId string
	client  ApicatRemoteAggregationServiceClient
	total   int64
	//用于接收来自汇聚端的消息
	notifier platform.Notifier
	ctx      context.Context
}

func NewGrpcClient(addr string, ctx context.Context, notifiers ...platform.Notifier) (*GrpcClient, error) {
	var c = &GrpcClient{
		ctx: ctx,
	}
	localId, err := machineid.ProtectedID("apicat")
	if err == nil {
		if len(notifiers) > 0 {
			c.notifier = notifiers[0]
		}
		c.localId = localId
		c.Connect(addr, ctx)
		return c, nil
	}
	return nil, err
}

func (g *GrpcClient) GetTotal() int64 {
	return g.total
}

func (g *GrpcClient) GetDetail() string {
	return fmt.Sprintf("send: %d\n", g.total)
}

var ErrNoNeedLoopInReady = errors.New("不需要在Ready中运行循环")

func (d *GrpcClient) RecordEntry(en *entry.LogEntry) error {
	var report = ToRemoteEntry(en, d.localId)
	d.total++
	_, err := d.client.Report(d.ctx, report)
	return err
}

func (d *GrpcClient) Connect(addr string, parent context.Context) error {
	ctx, _ := context.WithTimeout(parent, 30*time.Second)
	logrus.WithField("address", addr).Infoln("connect to server")
	conn, err := grpc.DialContext(ctx, addr, grpc.WithInsecure(), grpc.WithKeepaliveParams(keepalive.ClientParameters{Time: 1 * time.Minute}))

	if err != nil {
		logrus.Errorf("Unable to connect to grpc server (%s) by ER (%v)", addr, err)
		return err
	} else {
		d.conn = conn
	}

	var state = d.conn.GetState()

	logrus.Info("current state is ", state)

	go d.CreateClient(conn)

	var readyC = make(chan bool)
	go d.waitForTimeOut(state, readyC)
	select {
	case <-ctx.Done():
		return errors.New("连接未就绪")
	case <-readyC:
		logrus.Infoln("grpc ready")
		return nil
	}
}

// func (d *GrpcClient) ReConnect() error {
// 	d.conn.GetState()
// 	d.conn.Close()
// 	return d.Connect()
// }

func (d *GrpcClient) waitForTimeOut(state connectivity.State, readyC chan bool) {
	ctx, cancel := context.WithCancel(d.ctx)

	var failureCount = 0

	for {
		notDone := d.conn.WaitForStateChange(ctx, state)

		if notDone {

			newState := d.conn.GetState()

			logrus.WithField("goroutine", getGoRoutineInfo()).WithField("old", state).WithField("new", newState).Errorln("state change")

			if newState == connectivity.Connecting {
				if cancel != nil {
					cancel()
				}
				ctx, cancel = context.WithCancel(d.ctx)
			} else if newState == connectivity.Ready {
				readyC <- true
				if newState != state {
					failureCount = 0
					go func() {
						for {
							err := d.LoopInReady(ctx)
							caused := errors.Cause(err)
							code := status.Code(caused)
							if caused == io.EOF || code == codes.Canceled || code == codes.Unavailable { //
								if cancel != nil {
									cancel()
									ctx, cancel = context.WithCancel(d.ctx)
								}
								return
							} else if err != ErrNoNeedLoopInReady {
								logrus.Errorf("keep loop in ready(err:%v)\n", err)
							} else {
								break
							}
						}
					}()
				}
			} else if state == connectivity.TransientFailure {
				//check network if too many
				failureCount++
				if failureCount > 50 {
					logrus.Error("too many failure, exit")
					cancel()
					break
				}

			}
			state = newState
		} else {
			logrus.Infoln("grpc loop exit")
			cancel()
			break
		}
	}
}

func (g *GrpcClient) CreateClient(conn *grpc.ClientConn) error {
	g.client = NewApicatRemoteAggregationServiceClient(conn)
	return nil
}

func getGoRoutineInfo() string {
	b := make([]byte, 64)
	b = b[:runtime.Stack(b, false)]
	return strings.Split(string(b), "[")[0]
}

func (g *GrpcClient) LoopInReady(ctx context.Context) error {
	orderClient, err := g.client.Order(ctx, &MachineInfo{
		Id: g.localId,
	})
	if err == nil {
		err = g.WaitForOrder(ctx, orderClient)
	}
	return err
}

func (g *GrpcClient) WaitForOrder(ctx context.Context, client ApicatRemoteAggregationService_OrderClient) error {
	// var back = context.Background()
	logrus.Info("开始收到请求")
	for {
		cmd, err := client.Recv()
		if err == nil {
			logrus.Info("收到请求", cmd)
			go func(cmd *Command) {
				var err error
				var res = cmd.Entry.ToLogEntryResponse()
				switch cmd.Type {
				case Type_TypeAddEntry:
					err = g.notifier.NotifyAdd(res.Entry)
				case Type_TypeDelEntry:
					err = g.notifier.NotifyDel(res.Entry)
				}
				if err != nil {
					logrus.WithError(err).Info("汇聚机返回错误")
				}
			}(cmd)
		}
		if err != nil {
			if status.Code(err) == codes.Canceled {
				logrus.Errorln("汇聚机连接结束")
			}
			return err
		}
	}
}
